package com.upcoming.data.storm.caller;

import com.upcoming.data.storm.util.MyUtil;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;

/**
 * 作者：tianjiayuan
 * 创建时间：2018-05-17 11:44
 * 类描述：
 * 修改人：
 * 修改时间：
 */
public class FakeCallLogReaderSpout implements IRichSpout {

    //收集器
    private SpoutOutputCollector spoutOutputCollector;

    //环境信息
    private TopologyContext context;

    private boolean completed = false;

    private Random randomGenerator = new Random();

    private Integer idx = 0;

    public FakeCallLogReaderSpout() {
        MyUtil.outLog2NC(this, "new FakeCallLogReaderSpout()");
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("index", "from", "to", "duration"));
    }

    public Map<String, Object> getComponentConfiguration() {
        return null;
    }

    /**
     * 初始化
     * @param map
     * @param topologyContext
     * @param spoutOutputCollector
     */
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        MyUtil.outLog2NC(this, "SPout.open()");
        this.spoutOutputCollector = spoutOutputCollector;
        this.context = topologyContext;
        List<Integer> taskIds = topologyContext.getComponentTasks("call-log-creator-bolt");
        for (Integer i : taskIds) {
            System.out.println("---------------->" + i+ "\n");
        }
    }

    public void close() {

        MyUtil.outLog2NC(this, "SPout.close()");
    }

    public void activate() {
        MyUtil.outLog2NC(this, "SPout.activate()");
    }

    public void deactivate() {
        MyUtil.outLog2NC(this, "SPout.deactivate()");
    }

    /**
     *
     */
    public void nextTuple() {
        if (idx <= 100) {
            List<String> mobileList = new ArrayList<String>();
            mobileList.add("1234123401");
            mobileList.add("1234123402");
            mobileList.add("1234123403");
            mobileList.add("1234123404");
            Integer localIdx = 0;
            while (localIdx++ < 10 && this.idx++ < 10) {
                String fromMobileNumber = mobileList.get(randomGenerator.nextInt(4));
                String toMobileNumber = mobileList.get(randomGenerator.nextInt(4));
                while (fromMobileNumber == toMobileNumber) {
                    toMobileNumber = mobileList.get(randomGenerator.nextInt(4));
                }
                Integer duration = randomGenerator.nextInt(60);
                int index = this.idx;
                MyUtil.outLog2NC(this, "nextTuple:" + index + ":" + fromMobileNumber);
                //输出
//                int index = getIndexFromZK();
                this.spoutOutputCollector.emit(new Values(index, fromMobileNumber, toMobileNumber, duration));
            }
        }
    }

    public void ack(Object msgId) {
        MyUtil.outLog2NC(this, "SPout.ack:" + msgId);
    }

    public void fail(Object o) {

    }


    /*public int getIndexFromZK() {
        try {
            ZooKeeper zk = new ZooKeeper("127.0.0.1:2181", 1000, new Watcher() {
                public void process(WatchedEvent watchedEvent) {
                }
            });
            Stat s = new Stat();
            byte[] b = zk.getData("/index", null, s);
            int i = 0;
            if (b == null) {
                i = 1;
                zk.setData("/index", MyUtil.int2byte(1), s.getVersion());
            } else {
                i = MyUtil.byte2int(b);
                zk.setData("/index", MyUtil.int2byte(i + 1), s.getVersion());
            }
            return i;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return -1;
    }*/
}
